/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#pragma once


#include <kmhdfs/client/file_status.h>
#include <kmhdfs/client/permission.h>
#include <kmhdfs/proto/client_datanode_protocol.pb.h>
#include <kmhdfs/proto/client_namenode_protocol.pb.h>
#include <kmhdfs/server/datanode_info.h>
#include <kmhdfs/common/exception.h>
#include <kmhdfs/common/exception_internal.h>
#include <kmhdfs/server/extended_block.h>
#include <kmhdfs/server/located_block.h>
#include <kmhdfs/server/located_blocks.h>
#include <kmhdfs/common/stack_printer.h>
#include <kmhdfs/client/ec_policy.h>
#include <kmhdfs/client/system_ec_policies.h>

#include <algorithm>
#include <cassert>

using namespace google::protobuf;

namespace Hdfs {
    namespace Internal {
        class Nothing {
        };

        template<typename T1 = Nothing, typename T2 = Nothing, typename T3 = Nothing,
            typename T4 = Nothing, typename T5 = Nothing, typename T6 = Nothing,
            typename T7 = Nothing, typename T8 = Nothing, typename T9 = Nothing,
            typename T10 = Nothing, typename T11 = Nothing>
        class UnWrapper : public UnWrapper<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, Nothing> {
        private:
            typedef UnWrapper<T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, Nothing> BaseType;

        public:
            UnWrapper(const HdfsRpcServerException &e) : BaseType(e), e(e) {
            }

            void TURBO_NORETURN TURBO_NOINLINE unwrap(const char *file,
                                                      int line) {
                if (e.getErrClass() == T1::ReflexName) {
                    throw T1(e.getErrMsg(), SkipPathPrefix(file), line, print_stack(1, STACK_DEPTH).c_str());
                } else {
                    BaseType::unwrap(file, line);
                }
            }

        private:
            const HdfsRpcServerException &e;
        };

        template<>
        class UnWrapper<Nothing, Nothing, Nothing, Nothing, Nothing, Nothing, Nothing,
                    Nothing, Nothing, Nothing, Nothing> {
        public:
            UnWrapper(const HdfsRpcServerException &e) : e(e) {
            }

            TURBO_NORETURN TURBO_NOINLINE void unwrap(const char *file,
                                                      int line) {
                THROW(HdfsIOException,
                      "Unexpected exception: when unwrap the rpc remote exception \"%s\", %s in %s: %d",
                      e.getErrClass().c_str(), e.getErrMsg().c_str(), file, line);
            }

        private:
            const HdfsRpcServerException &e;
        };

        static inline void Convert(ExtendedBlock &eb,
                                   const ExtendedBlockProto &proto) {
            eb.setBlockId(proto.blockid());
            eb.setGenerationStamp(proto.generationstamp());
            eb.setNumBytes(proto.numbytes());
            eb.setPoolId(proto.poolid());
        }

        static inline void Convert(Token &token,
                                   const TokenProto &proto) {
            token.setIdentifier(proto.identifier());
            token.setKind(proto.kind());
            token.setPassword(proto.password());
            token.setService(proto.service());
        }

        static inline void Convert(DatanodeInfo &node,
                                   const DatanodeInfoProto &proto) {
            const DatanodeIDProto &idProto = proto.id();
            node.setHostName(idProto.hostname());
            node.setInfoPort(idProto.infoport());
            node.setIpAddr(idProto.ipaddr());
            node.setIpcPort(idProto.ipcport());
            node.setDatanodeId(idProto.datanodeuuid());
            node.setXferPort(idProto.xferport());
            node.setLocation(proto.location());
        }

        static inline std::shared_ptr<LocatedBlock> Convert(const LocatedBlockProto &proto) {
            std::shared_ptr<LocatedBlock> lb(new LocatedBlock);
            if (proto.has_blockindices()) {
                lb->setStriped(true);
                std::vector<int8_t> &indeces = lb->mutableIndices();
                indeces.resize(proto.blockindices().size());
                for (int i = 0; i < static_cast<int>(proto.blockindices().size()); ++i) {
                    indeces[i] = ((int8_t) proto.blockindices()[i]);
                }
                std::vector<Token> &tokens = lb->mutableTokens();
                tokens.resize(proto.blocktokens_size());
                for (int i = 0; i < proto.blocktokens_size(); ++i) {
                    Token token;
                    Convert(token, proto.blocktokens(i));
                    tokens[i] = token;
                }
            }
            Token token;
            Convert(token, proto.blocktoken());
            lb->setToken(token);
            std::vector<DatanodeInfo> &nodes = lb->mutableLocations();
            nodes.resize(proto.locs_size());

            for (int i = 0; i < proto.locs_size(); ++i) {
                Convert(nodes[i], proto.locs(i));
            }

            if (proto.storagetypes_size() > 0) {
                assert(proto.storagetypes_size() == proto.locs_size());
                std::vector<std::string> &storageIDs = lb->mutableStorageIDs();
                storageIDs.resize(proto.storagetypes_size());

                for (int i = 0; i < proto.storagetypes_size(); ++i) {
                    storageIDs[i] = proto.storageids(i);
                }
            }

            Convert(*lb, proto.b());
            lb->setOffset(proto.offset());
            lb->setCorrupt(proto.corrupt());
            return lb;
        }

        static inline void Convert(LocatedBlocks &lbs,
                                   const ErasureCodingPolicyProto &proto) {
            int8_t id = (int8_t) (proto.id() & 0xFF);
            SystemECPolicies &sysPolicy = SystemECPolicies::getInstance();
            std::shared_ptr<ECPolicy> policy = sysPolicy.getById(id);
            if (policy != nullptr) {
                lbs.setEcPolicy(policy);
            } else {
                policy = std::shared_ptr<ECPolicy>(new ECPolicy());
                policy->setId(id);
                policy->setName(proto.name().c_str());
                policy->setCellSize(proto.cellsize());
                policy->setCodecName(proto.schema().codecname().c_str());
                policy->setNumDataUnits(proto.schema().dataunits());
                policy->setNumParityUnits(proto.schema().parityunits());
                sysPolicy.addEcPolicy(proto.id(), policy);
                lbs.setEcPolicy(policy);
            }
        }

        static inline void Convert(LocatedBlocks &lbs,
                                   const LocatedBlocksProto &proto) {
            std::shared_ptr<LocatedBlock> lb;
            lbs.setFileLength(proto.filelength());
            lbs.setIsLastBlockComplete(proto.islastblockcomplete());
            lbs.setUnderConstruction(proto.underconstruction());

            if (proto.has_lastblock()) {
                lb = Convert(proto.lastblock());
                lbs.setLastBlock(lb);
            }

            std::vector<LocatedBlock> &blocks = lbs.getBlocks();
            blocks.resize(proto.blocks_size());

            for (int i = 0; i < proto.blocks_size(); ++i) {
                blocks[i] = *Convert(proto.blocks(i));
            }

            std::sort(blocks.begin(), blocks.end(), std::less<LocatedBlock>());
            if (proto.has_ecpolicy()) {
                Convert(lbs, proto.ecpolicy());
            }
        }

        static inline void Convert(FileStatus &fs,
                                   const ErasureCodingPolicyProto &proto) {
            int8_t id = (int8_t) (proto.id() & 0xFF);
            SystemECPolicies &sysPolicy = SystemECPolicies::getInstance();
            std::shared_ptr<ECPolicy> policy = sysPolicy.getById(id);
            if (policy != nullptr) {
                fs.setEcPolicy(policy);
            } else {
                policy = std::shared_ptr<ECPolicy>(new ECPolicy());
                policy->setId(id);
                policy->setName(proto.name().c_str());
                policy->setCellSize(proto.cellsize());
                policy->setCodecName(proto.schema().codecname().c_str());
                policy->setNumDataUnits(proto.schema().dataunits());
                policy->setNumParityUnits(proto.schema().parityunits());
                sysPolicy.addEcPolicy(proto.id(), policy);
                fs.setEcPolicy(policy);
            }
        }

        static inline void Convert(const std::string &src, FileStatus &fs,
                                   const HdfsFileStatusProto &proto) {
            fs.setAccessTime(proto.access_time());
            fs.setBlocksize(proto.blocksize());
            fs.setGroup(proto.group().c_str());
            fs.setLength(proto.length());
            fs.setModificationTime(proto.modification_time());
            fs.setOwner(proto.owner().c_str());
            fs.setPath((src + "/" + proto.path()).c_str());
            fs.setReplication(proto.block_replication());
            fs.setSymlink(proto.symlink().c_str());
            fs.setPermission(Permission(proto.permission().perm()));
            fs.setIsdir(proto.filetype() == HdfsFileStatusProto::IS_DIR);
            fs.setFileId(proto.fileid());
            if (proto.has_ecpolicy()) {
                Convert(fs, proto.ecpolicy());
            }
        }

        static inline void Convert(const std::string &src,
                                   std::vector<FileStatus> &dl,
                                   const DirectoryListingProto &proto) {
            RepeatedPtrField<HdfsFileStatusProto> ptrproto = proto.partiallisting();

            for (int i = 0; i < ptrproto.size(); i++) {
                FileStatus fileStatus;
                Convert(src, fileStatus, ptrproto.Get(i));
                dl.push_back(fileStatus);
            }
        }

        static inline Token Convert(const TokenProto &proto) {
            Token retval;
            retval.setIdentifier(proto.identifier());
            retval.setKind(proto.kind());
            retval.setPassword(proto.password());
            return retval;
        }

        /*static inline void Convert(ContentSummary & contentSummary, const ContentSummaryProto & proto) {
            contentSummary.setDirectoryCount(proto.directorycount());
            contentSummary.setFileCount(proto.filecount());
            contentSummary.setLength(proto.length());
            contentSummary.setQuota(proto.quota());
            contentSummary.setSpaceConsumed(proto.spaceconsumed());
            contentSummary.setSpaceQuota(proto.spacequota());
        }*/

        static inline void Build(const Token &token,
                                 TokenProto *proto) {
            proto->set_identifier(token.getIdentifier());
            proto->set_kind(token.getKind());
            proto->set_password(token.getPassword());
            proto->set_service(token.getService());
        }

        static inline void Build(const Permission &p, FsPermissionProto *proto) {
            proto->set_perm(p.toShort());
        }

        static inline void Build(const DatanodeInfo &dn, DatanodeIDProto *proto) {
            proto->set_hostname(dn.getHostName());
            proto->set_infoport(dn.getInfoPort());
            proto->set_ipaddr(dn.getIpAddr());
            proto->set_ipcport(dn.getIpcPort());
            proto->set_datanodeuuid(dn.getDatanodeId());
            proto->set_xferport(dn.getXferPort());
        }

        static inline void Build(const std::vector<DatanodeInfo> &dns,
                                 RepeatedPtrField<DatanodeInfoProto> *proto) {
            for (size_t i = 0; i < dns.size(); ++i) {
                DatanodeInfoProto *p = proto->Add();
                Build(dns[i], p->mutable_id());
                p->set_location(dns[i].getLocation());
            }
        }

        static inline void Build(const ExtendedBlock &eb, ExtendedBlockProto *proto) {
            proto->set_blockid(eb.getBlockId());
            proto->set_generationstamp(eb.getGenerationStamp());
            proto->set_numbytes(eb.getNumBytes());
            proto->set_poolid(eb.getPoolId());
        }

        static inline void Build(LocatedBlock &b, LocatedBlockProto *proto) {
            proto->set_corrupt(b.isCorrupt());
            proto->set_offset(b.getOffset());
            Build(b, proto->mutable_b());
            Build(b.getLocations(), proto->mutable_locs());
        }

        /*static inline void Build(const std::vector<LocatedBlock> & blocks,
                                 RepeatedPtrField<LocatedBlockProto> * proto) {
            for (size_t i = 0; i < blocks.size(); ++i) {
                LocatedBlockProto * p = proto->Add();
                p->set_corrupt(blocks[i].isCorrupt());
                p->set_offset(blocks[i].getOffset());
                Build(blocks[i], p->mutable_b());
            }
        }*/

        static inline void Build(const std::vector<std::string> &srcs,
                                 RepeatedPtrField<std::string> *proto) {
            for (size_t i = 0; i < srcs.size(); ++i) {
                proto->Add()->assign(srcs[i]);
            }
        }

        static inline void Build(const std::vector<DatanodeInfo> &dns,
                                 RepeatedPtrField<DatanodeIDProto> *proto) {
            for (size_t i = 0; i < dns.size(); ++i) {
                Build(dns[i], proto->Add());
            }
        }
    }
}
